/**  
 * @Title: FlowTaskService.java
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-09-12 13:40:59 
 */  
package com.fsxgt.datagrid.executor.service;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.PostConstruct;

import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson2.JSON;
import com.fsxgt.datagrid.component.bo.CmpSyncOfflineBO;
import com.fsxgt.datagrid.component.bo.CmpSyncOfflineColumnBO;
import com.fsxgt.datagrid.component.bo.DataWorkBO;
import com.fsxgt.datagrid.component.bo.DataWorkColumnBO;
import com.fsxgt.datagrid.component.service.impl.DataWorkService;
import com.fsxgt.datagrid.core.component.HttpComponent;
import com.fsxgt.datagrid.core.exception.DataGridException;
import com.fsxgt.datagrid.core.task.TaskRunParam;
import com.fsxgt.datagrid.core.task.SubTask;
import com.fsxgt.datagrid.core.task.Task;
import com.fsxgt.datagrid.core.task.TaskCallback;
import com.fsxgt.datagrid.core.task.TaskContainer;
import com.fsxgt.datagrid.core.task.TaskContext;
import com.fsxgt.datagrid.core.utils.LogHelpper;
import com.fsxgt.datagrid.core.utils.ReturnT;
import com.fsxgt.datagrid.ds.utils.ParamUtil;
import com.fsxgt.datagrid.flow.handler.FlowContext;
import com.fsxgt.datagrid.flow.handler.FlowReturn;
import com.fsxgt.datagrid.flow.handler.IComponentHandler;
import com.fsxgt.datagrid.flow.service.impl.CmpRegistryService;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**  
 * @ClassName: FlowTaskService
 * @Description: TODO(描述)
 * @author liuyi
 * @date 2023-09-12 13:40:59 
*/
@Slf4j
@Service
public class DataWorkTaskService {



	@Autowired
	private DataWorkService dataWorkService;
	
	@Autowired
	private CmpRegistryService cmpRegistryService;
	

	
	@Autowired
	private HttpComponent httpComponent;

	@Value("${console.address}")
	private String console;
	
	@Value("${executor.main.thread.threadSize}")
	private int threadSize;
	
	private TaskContainer taskContainer;


	@PostConstruct
	private void init() {
		taskContainer=taskContainer.getInstance(threadSize);
	}

	
	
	@Data
	private class DataWorkTask extends Task{

		private DataWorkBO dataWorkBO;
		private String version;
		
		
		public DataWorkTask(String id,DataWorkBO dataWorkBO,String version) {
			super(id,dataWorkBO.getName());
			this.dataWorkBO=dataWorkBO;
			this.version=version;
		}
		
	}
	
	private Map<String, DataWorkTask> cacheMap = new ConcurrentHashMap();
	
	public Task getTask(String dataWorkId,String version) throws DataGridException, ParseException {
		
		DataWorkTask dataWorkTask=null;
		

		if (version != null && cacheMap.containsKey(dataWorkId)&&cacheMap.get(dataWorkId).getVersion()!=null
				&& cacheMap.get(dataWorkId).getVersion().equals(version)) {
			// 使用缓存
			dataWorkTask=cacheMap.get(dataWorkId);
		} else {
			DataWorkBO dataWorkBO = dataWorkService.getDataWork(dataWorkId);
			
			dataWorkTask=new DataWorkTask(dataWorkBO.getId(),dataWorkBO,version);
			
			CmpSyncOfflineBO cmpSyncOfflineBO=new CmpSyncOfflineBO();
			BeanUtils.copyProperties(dataWorkBO, cmpSyncOfflineBO);
			
			for(DataWorkColumnBO column:dataWorkBO.getColumnList()) {
				CmpSyncOfflineColumnBO cmpSyncOfflineColumnBO=new CmpSyncOfflineColumnBO();
				BeanUtils.copyProperties(column, cmpSyncOfflineColumnBO);
				cmpSyncOfflineBO.getColumnList().add(cmpSyncOfflineColumnBO);
			}
			 
			
			IComponentHandler handler = cmpRegistryService.getHandler("Data Work");


			dataWorkTask.addSubTask(new SubTask("0",dataWorkBO.getName()) {
				@Override
				public FlowReturn handler(TaskContext context) throws Exception {
					// TODO Auto-generated method stub
					return handler.handler(cmpSyncOfflineBO,(FlowContext) context);
				}

				@Override
				public void stopHandler(TaskContext context) throws Exception{
					// TODO Auto-generated method stub
					handler.stopHandler(cmpSyncOfflineBO,(FlowContext) context);
				}
			});
			
			cacheMap.put(dataWorkId, dataWorkTask);
		}
		return dataWorkTask;
		
	}
	
	public TaskContext getContext() {

		return FlowTaskService.getContext();
		
	}
	
	public String execute(String dataWorkId, Date logDate, String execId, String version,Integer threads,Integer timeout,Integer retryCount,Map context1) throws DataGridException, ParseException {
		
		
		Task task=getTask(dataWorkId, version);
		TaskContext context=getContext();
		
		Map p=(Map) context.get(ParamUtil.PARAM_KEY);
		
		if(context1!=null) {
			context1.forEach((k,v)->{
				p.put(k.toString(), v);
			});
		}
		
		String logFileName = LogHelpper.makeLogFileName(logDate, execId);
		
		TaskRunParam param=new TaskRunParam(execId,logFileName,threads,timeout,retryCount);

		taskContainer.execute(task,context,param,new TaskCallback() {
			@Override
			public void handler(Date startTime, Date overTime, String status, String logPath, String execId) {
				// TODO Auto-generated method stub
				SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

				Map param = new HashMap();
				param.put("overTime", format.format(overTime));
				param.put("startTime", format.format(startTime));
				param.put("status", status);
				param.put("logPath", logPath);
				param.put("execId", execId);
				param.put("objType", "2");
				param.put("objId", dataWorkId);

				log.info("callback:" + "http://" + console + "/executor/callback");
				log.info("callback:" + JSON.toJSONString(param));

				ReturnT<Map> result = httpComponent.sendHttpPostRetry("http://" + console + "/executor/callback", param,
						ReturnT.class);
			}
		});
		
		
		
		return execId;
	}

	/**  
	 * @Title: interrupted
	 * @Description: TODO(描述)
	 * @param string
	 * @author liuyi
	 * @date 2023-09-12 15:17:19 
	 */  
	public void interrupted(String execId) {
		// TODO Auto-generated method stub
		taskContainer.interrupted(execId);
	}
	
	
}
